package com.nenglai.common.task;

import com.nenglai.common.util.SpringUtil;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.SerializationUtils;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

public class TaskUtil {

    private static final Logger logger = LoggerFactory.getLogger(TaskUtil.class);

    @Setter
    private ITaskSuccess taskSuccess = new TaskSuccess();

    @Setter
    private ITaskFail taskFail = new TaskFail();

    @Setter
    private ITaskRetry taskRetry = new TaskRetry();

    /**
     * 增长id
     */
    private final AtomicInteger counter = new AtomicInteger(0);

    /**
     * 任务并发数
     */
    private Integer poolSize = 5;

    /**
     * 队列长度
     */
    private Integer dequeCapacity = 10000;

    /**
     * 队列池
     */
    private Map<Integer, BlockingDeque<Message>> blockingDequePool;

    public TaskUtil(Integer poolSize, Integer dequeCapacity) {
        this.poolSize = poolSize;
        this.dequeCapacity = dequeCapacity;
    }

    /**
     * 启动队列监听
     *
     * @return void
     * @author kongdehui
     * @date 2021/05/19 15:06
     */
    public void start() {
        // 1、初始化阻塞队列
        blockingDequePool = new ConcurrentHashMap<>(poolSize);
        for (int i = 0; i < poolSize; i++) {
            blockingDequePool.put(i, new LinkedBlockingDeque<>(dequeCapacity));
        }

        // 2、监听阻塞队列
        Set<Map.Entry<Integer, BlockingDeque<Message>>> entries = blockingDequePool.entrySet();
        for (Map.Entry<Integer, BlockingDeque<Message>> entry : entries) {
            new Thread(() -> {
                try {
                    for (; ; ) {
                        Message message = entry.getValue().take();
                        try {
                            int state = invoke(message);

                            if (state == TaskEnum.INVOKE_SUCCESS.getCode()) {
                                transToSuccess(message);
                                continue;
                            }

                            if (state == TaskEnum.INVOKE_FAIL.getCode()) {
                                transToError(message);
                                continue;
                            }

                            if (state == TaskEnum.INVOKE_RETRY.getCode()) {
                                transToRetry(message);
                                message.setTimes(message.getTimes() + 1);
                                message.setRemark("重试次数=" + (message.getTimes() - 1));
                                setMessage(message);
                            }

                        } catch (Exception e) {
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
            logger.info("...启动Task处理器{}", entry.getKey());
        }

    }


    /**
     * 0 不处理
     * 1 成功
     * 2 失败
     * 3 重试
     *
     * @param message
     * @return
     */
    private int invoke(Message message) {
        if (message == null) {
            return TaskEnum.INVOKE_NONE.getCode();
        }

        ITaskHeader header = null;

        try {

            header = (ITaskHeader) SerializationUtils.deserialize(message.getHeader());
            ITaskBody body = (ITaskBody) SerializationUtils.deserialize(message.getBody());
            Class<?> clazz = Class.forName(message.getProcessor());
            ITask task = (ITask) SpringUtil.getBean(clazz);
            task.execute(header, body);
            return TaskEnum.INVOKE_SUCCESS.getCode();

        } catch (Exception e) {
            if (header != null) {
                int times = header.retryTimes();
                if (times >= message.getTimes()) {
                    logger.error("[task]processor={}", message.getProcessor(), e);
                    return TaskEnum.INVOKE_RETRY.getCode();
                }
            }

            e.printStackTrace();
        }

        return TaskEnum.INVOKE_FAIL.getCode();
    }

    /**
     * 投递消息
     *
     * @param message
     */
    private boolean setMessage(Message message) {
        Integer taskIndex = message.getTaskIndex();
        int key;
        if (taskIndex != null && blockingDequePool.containsKey(taskIndex)) {
            key = taskIndex;
        } else {

            // 通过自定义池
            List<Integer> pools = message.getPools();
            if (pools != null && pools.size() > 0) {
                key = message.getId() % pools.size();
                Integer i = pools.get(key);
                return blockingDequePool.get(i).add(message);
            }

            key = message.getId() % poolSize;
        }

        if (blockingDequePool.get(key).size() >= dequeCapacity) {
            return false;
        }

        return blockingDequePool.get(key).add(message);
    }

    public boolean push(ITaskBody body, Class<? extends ITask<? extends ITaskHeader, ? extends ITaskBody>> clazz) {
        return push(new DefaultHeader(), body, clazz, null, null);
    }

    public boolean push(ITaskHeader header,
                        ITaskBody body,
                        Class<? extends ITask<? extends ITaskHeader, ? extends ITaskBody>> clazz) {
        return push(header, body, clazz, null, null);
    }

    public boolean push(ITaskHeader header,
                        ITaskBody body,
                        Class<? extends ITask<? extends ITaskHeader, ? extends ITaskBody>> clazz,
                        Integer index,
                        List<Integer> pools) {
        String name = clazz.getName();
        Message messageQueue = new Message();

        int id = counter.incrementAndGet();
        if (id >= Integer.MAX_VALUE) {
            counter.set(0);
        }
        messageQueue.setId(id);
        messageQueue.setTaskIndex(index);
        messageQueue.setHeader(SerializationUtils.serialize(header));
        messageQueue.setBody(SerializationUtils.serialize(body));
        messageQueue.setProcessor(name);
        messageQueue.setCreateTime(new Date());
        messageQueue.setUpdateTime(new Date());
        messageQueue.setState(TaskEnum.STATE0.getCode());
        messageQueue.setTimes(1);
        messageQueue.setPools(pools);
        return setMessage(messageQueue);
    }

    public void printTask() {
        logger.info("======[task]pool={}======", poolSize);
        for (int i = 0; i < poolSize; i++) {
            logger.info("队列{}，任务数={}", i, blockingDequePool.get(i).size());
        }
        logger.info("==========================");
    }

    /**
     * @param message
     */
    public void transToSuccess(Message message) {
        taskSuccess.success(message);
    }

    /**
     * @param message
     */
    public void transToError(Message message) {
        taskFail.error(message);
    }

    private void transToRetry(Message message) {
        taskRetry.retry(message);
    }

    static class TaskSuccess implements ITaskSuccess {
        @Override
        public void success(Message message) {
            logger.info("[task]任务：{}，处理成功！", message.getProcessor());
        }
    }

    static class TaskFail implements ITaskFail {
        @Override
        public void error(Message message) {
            logger.error("[task]任务：{}，处理失败！", message);
        }
    }

    static class TaskRetry implements ITaskRetry {
        @Override
        public void retry(Message message) {
            logger.info("[task]processor={} 重试第{}次！", message.getProcessor(), message.getTimes());
        }
    }

}
